package worker

import (
	"context"
	"duanju-mq/pkg/app"
	"duanju-mq/pkg/config"
	"sync"
	"time"
)

var (
	fBuffer  = config.Int("lib.worker.dispatcher.buffer", 64, "")
	fIdleSec = config.Int("lib.worker.dispatcher.idlesec", 0, "")
)

type Dispatcher[T any] struct {
	data    map[string]*worker[T]
	lock    sync.Mutex
	handler func(T)
	buflen  int
	idlesec int
}

func NewDispatcher[T any](handler func(T)) *Dispatcher[T] {
	return &Dispatcher[T]{
		data:    map[string]*worker[T]{},
		handler: handler,
		buflen:  *fBuffer,
		idlesec: *fIdleSec,
	}
}

func (d *Dispatcher[T]) SetBufLen(n int)  { d.buflen = n }
func (d *Dispatcher[T]) SetIdleSec(n int) { d.idlesec = n }

func (d *Dispatcher[T]) Handle(name string, msg T) {
	w := d.getWorker(name)
	w.ch <- msg
}

func (d *Dispatcher[T]) getWorker(name string) *worker[T] {
	d.lock.Lock()
	defer d.lock.Unlock()
	if w := d.data[name]; w != nil {
		return w
	}
	w := &worker[T]{d: d, name: name, ch: make(chan T, d.buflen), atime: time.Now().Unix()}
	w.ctx, w.cancel = context.WithCancel(app.CTX)
	app.AsyncSafe(w.run)
	app.Async(w.checkIdle)
	d.data[name] = w
	return w
}

func (d *Dispatcher[T]) removeWorker(name string) {
	d.lock.Lock()
	defer d.lock.Unlock()
	delete(d.data, name)
}

type worker[T any] struct {
	d      *Dispatcher[T]
	name   string
	ch     chan T
	ctx    context.Context
	cancel context.CancelFunc
	atime  int64
}

func (w *worker[T]) run() {
	defer w.d.removeWorker(w.name)
	for {
		select {
		case <-w.ctx.Done():
			goto remainder
		case msg := <-w.ch:
			w.atime = time.Now().Unix()
			w.d.handler(msg)
		}
	}
remainder:
	for {
		select {
		case msg := <-w.ch:
			w.atime = time.Now().Unix()
			w.d.handler(msg)
		default:
			return
		}
	}
}

func (w *worker[T]) checkIdle() {
	defer w.cancel()
	for w.ctx.Err() == nil {
		time.Sleep(time.Second)
		if w.d.idlesec > 0 && time.Now().Unix()-w.atime >= int64(w.d.idlesec) {
			return
		}
	}
}
